package net.bwie.realtime.job;

import com.alibaba.fastjson.JSON;
import net.bwie.realtime.dws.log.bean.MonitorInfo;
import net.bwie.realtime.dws.log.bean.Violation;
import net.bwie.realtime.dws.log.function.TaoPaiRichFlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import utils.DorisUtil;
import utils.KafkaUtil;

public class TaoPaiCarController {

    public static void main(String[] args) throws Exception{
        //1.执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取数据源
        DataStream<String> carStream = KafkaUtil.consumerKafka(env, "traffic_monitor_data");
//        carStream.print();

        //3.数据转换
        DataStream<String> resultStream = handle(carStream);
        resultStream.print("result");

        //4.数据输出
//        DorisUtil.saveToDoris(resultStream,"","");



        //5.执行
        env.execute("TaoPaiCarController");
    }

    /**
     * 套牌分析:当某个卡口中出现一辆行驶的汽车，我们可以通过摄像头识别车牌号，然后在10秒内，
     * 另外一个卡口（或者当前卡口）也识别到了同样车牌的车辆，那么很有可能这两辆车之中有很大几率存在套牌车，
     * 因为一般情况下不可能有车辆在10秒内经过两个卡口。如果发现涉嫌套牌车，系统实时发出报警信息，同时这些存在套牌车嫌疑的车辆
     * @param stream
     * @return
     */
    private static DataStream<String> handle(DataStream<String> stream) {
        //a.将数据封装成bean对象
        SingleOutputStreamOperator<MonitorInfo> mapStream = stream.map(
                (MapFunction<String, MonitorInfo>) value -> {
                    MonitorInfo monitorInfo = JSON.parseObject(value, MonitorInfo.class);
                    return monitorInfo;
                }
        );

        //b.数据处理转换
        SingleOutputStreamOperator<Violation> resultDs = mapStream.keyBy(
                speedInfo -> speedInfo.getCar()
        ).flatMap(new TaoPaiRichFlatMapFunction());

        //c.数据输出
        SingleOutputStreamOperator<String> resultStream = resultDs.map(
                (MapFunction<Violation, String>) value -> JSON.toJSONString(value)
        );

        return resultStream;
    }
}
